package cn.getech.data.development.service.impl;

import cn.getech.data.development.constant.DataDevelopmentBizExceptionEnum;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.mapper.*;
import cn.getech.data.development.model.dto.*;
import cn.getech.data.development.model.qo.ConfConnectQO;
import cn.getech.data.development.service.IAnalysisMessageModelConfService;
import cn.getech.data.development.service.ModelInfoService;
import cn.getech.data.development.utils.MyKafkaUtils;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.PojoUtils;
import cn.getech.system.center.utils.ShiroUtils;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>
 * 原始数据和模型的映射信息 服务实现类
 * </p>
 *
 * @author hujz
 * @since 2019-08-21
 */
@Service
public class AnalysisMessageModelConfServiceImpl extends ServiceImpl<AnalysisMessageModelConfMapper, AnalysisMessageModelConf> implements IAnalysisMessageModelConfService {

    @Autowired
    private AnalysisInfoMapper analysisInfoMapper;
    @Autowired
    private ModelInfoMapper modelInfoMapper;
    @Autowired
    private ConfConnectMapper confConnectMapper;
    @Autowired
    private ModelDetailsMapper modelDetailsMapper;

    @Autowired
    private AnalysisMessageModelMapper analysisMessageModelMapper;

    @Autowired
    private AnalysisMessageModelConfMapper analysisMessageModelConfMapper;

    @Autowired
    private TableInfoMapper tableInfoMapper;

    @Autowired
    private ModelInfoService modelInfoService;

    @Autowired
    private TableFieldInfoMapper tableFieldInfoMapper;


    @Override
    public AnalysisModelDataAndMappingDto getDataByConnect(AnalysisInfoDto analysisInfoDto) {

        AnalysisModelDataAndMappingDto analysisModelDataAndMappingDto = new AnalysisModelDataAndMappingDto();

        //验证参数
        cheackParam(analysisInfoDto);
        AnalysisInfo analysisInfo = analysisInfoMapper.selectById(analysisInfoDto.getId());
        if (null == analysisInfo) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "分析任务"));
        }

        //获取模型的关系数据
        ModelInfo modelInfo = modelInfoMapper.selectById(analysisInfoDto.getModelInfoId());
        if (null == modelInfo || Objects.equals(modelInfo.getIsDelete(), 1)) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "模型"));
        }
        //根据连接获取kafka中的数据
        ConfConnect confConnect = confConnectMapper.selectById(analysisInfoDto.getConnectId());
        if (null == confConnect) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "连接"));
        }
        //匹配模型中的数据
        List<ModelDetails> modelDetails = modelDetailsMapper.selectList(new QueryWrapper<ModelDetails>().eq("model_info_id", modelInfo.getId()).eq("model_type", 1));
        if (null == modelDetails || modelDetails.isEmpty()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "模型映射关系"));
        }

        //查找所有的连接
        ConfConnectQO confConnectQO = new ConfConnectQO();
        confConnectQO.setProcId(analysisInfo.getProcId());
        confConnectQO.setTypeId(6);
        List<ConfConnect> confConnects = confConnectMapper.selectListByTypeIdAndProceId(confConnectQO);
        JSONArray objects = modelInfoService.dbTableList(2);

        //之前是否配置过模型
        if (Objects.equals(analysisInfoDto.getConnectId(), analysisInfo.getConnectId()) && Objects.equals(analysisInfoDto.getModelInfoId(), analysisInfo.getModelInfoId())) {
            //根据任务id去查询是否已经配置过
            String msg = analysisInfo.getMessageData();
            if (StringUtils.isEmpty(msg)) {
                return analysisModelDataAndMappingDto;
            }
            //拼接数据
            pingResultData(analysisModelDataAndMappingDto, msg, modelDetails);

            //另外还得把配置的信息返回
            List<AnalysisInfoConfigDto> havechoiceAnalysisInfoConfigDtoList = analysisMessageModelConfMapper.selectListByAnalysisInfoId(analysisInfo.getId());
            analysisModelDataAndMappingDto.setAnalysisInfoConfigDtos(havechoiceAnalysisInfoConfigDtoList);
            analysisModelDataAndMappingDto.setConfConnectDtos(PojoUtils.listConvert(ConfConnectDto.class, confConnects));
            analysisModelDataAndMappingDto.setOutPutList(objects);
            return analysisModelDataAndMappingDto;

        }

        //编辑的验证，需要判断其状态为禁用状态
//        if (analysisInfo.getEnable()) {
//            throw new RRException("分析任务是启用状态，不能修改！");
//        }
        //验证参数
        cheackConfConnect(confConnect);
        //初始化kafka消费者
        ConfConnectDto confConnectDto = new ConfConnectDto();
        BeanUtils.copyProperties(confConnect, confConnectDto);
        //0_cofconnectId 只用来获取一笔数据而已
        confConnectDto.setAnalysisInfoId(0);
        String uuids = MyKafkaUtils.initKafKaConsumer(confConnectDto);
        //暂定5s之内获取数据
        String kafkaLastData = MyKafkaUtils.getKafkaLastData(5000, uuids + "_" + confConnectDto.getId());
        if (StringUtils.isEmpty(kafkaLastData)) {
            throw new RRException("消费kafka数据不存在");
        }
        pingResultData(analysisModelDataAndMappingDto, kafkaLastData, modelDetails);
        analysisModelDataAndMappingDto.setConfConnectDtos(PojoUtils.listConvert(ConfConnectDto.class, confConnects));
        analysisModelDataAndMappingDto.setOutPutList(objects);

        return analysisModelDataAndMappingDto;
    }

    private void cheackParam(AnalysisInfoDto analysisInfoDto) {
        if (null == analysisInfoDto || null == analysisInfoDto.getId()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_FAIL_ERROR.getMessage(), "分析任务"));
        }
        if (null == analysisInfoDto.getModelInfoId()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_FAIL_ERROR.getMessage(), "模型id"));
        }
        if (null == analysisInfoDto.getConnectId()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_FAIL_ERROR.getMessage(), "数据源连接"));
        }
    }

    @Override
    public AnalysisModelDataAndMappingDto reloadModelCof(AnalysisInfoDto analysisInfoDto) {

        AnalysisModelDataAndMappingDto analysisModelDataAndMappingDto = new AnalysisModelDataAndMappingDto();

        cheackParam(analysisInfoDto);
        AnalysisInfo analysisInfo = analysisInfoMapper.selectById(analysisInfoDto.getId());
        if (null == analysisInfo) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "分析任务"));
        }
        //编辑的验证，需要判断其状态为禁用状态
        if (analysisInfo.getEnable()) {
            throw new RRException("分析任务是启用状态，不能重新加载！");
        }
        //验证是否是重新加载的
        if (null == analysisInfo.getModelInfoId()) {
            throw new RRException("分析任务之前没有配置过,不能重新加载！");
        }
        if (null != analysisInfo.getModelInfoId()) {
            if (!Objects.equals(analysisInfoDto.getConnectId(), analysisInfo.getConnectId()) || !Objects.equals(analysisInfoDto.getModelInfoId(), analysisInfo.getModelInfoId())) {
                throw new RRException("分析任务是新的配置，无需重新加载！");
            }
        }

        //获取模型的关系数据
        ModelInfo modelInfo = modelInfoMapper.selectById(analysisInfoDto.getModelInfoId());
        if (null == modelInfo || Objects.equals(modelInfo.getIsDelete(), 1)) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "模型"));
        }
        //根据连接获取kafka中的数据
        ConfConnect confConnect = confConnectMapper.selectById(analysisInfoDto.getConnectId());
        if (null == confConnect) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "连接"));
        }
        //匹配模型中的数据
        List<ModelDetails> modelDetails = modelDetailsMapper.selectList(new QueryWrapper<ModelDetails>().eq("model_info_id", modelInfo.getId()).eq("model_type", 1));
        if (null == modelDetails || modelDetails.isEmpty()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "模型映射关系"));
        }
        //验证参数
        cheackConfConnect(confConnect);
        //初始化kafka消费者
        ConfConnectDto confConnectDto = new ConfConnectDto();
        BeanUtils.copyProperties(confConnect, confConnectDto);
        //0_cofconnectId 只用来获取一笔数据而已
        confConnectDto.setAnalysisInfoId(0);
        String uuids = MyKafkaUtils.initKafKaConsumer(confConnectDto);
        //暂定5s之内获取数据
        String kafkaLastData = MyKafkaUtils.getKafkaLastData(5000, uuids + "_" + confConnectDto.getId());
        if (StringUtils.isEmpty(kafkaLastData)) {
            throw new RRException("消费kafka数据不存在");
        }
        pingResultData(analysisModelDataAndMappingDto, kafkaLastData, modelDetails);

        return analysisModelDataAndMappingDto;


    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = RuntimeException.class)
    public void save(AnalysisMessageModelConfSaveDto analysisMessageModelConfSaveDto) {

        AnalysisInfo analysisInfo = analysisInfoMapper.selectById(analysisMessageModelConfSaveDto.getId());
        if (null == analysisInfo) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "分析任务"));
        }
        //验证是修改还是新增
        if (StringUtils.isNotEmpty(analysisInfo.getMessageData())) {
            if (analysisInfo.getEnable()) {
                throw new RRException("分析任务是启用状态，不能重新加载！");
            }
        }
        //获取模型的关系数据
        ModelInfo modelInfo = modelInfoMapper.selectById(analysisMessageModelConfSaveDto.getModelInfoId());
        if (null == modelInfo || Objects.equals(modelInfo.getIsDelete(), 1)) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "模型"));
        }
        //根据连接获取kafka中的数据
        ConfConnect confConnect = confConnectMapper.selectById(analysisMessageModelConfSaveDto.getConnectId());
        if (null == confConnect) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "连接"));
        }
        //匹配模型中的数据
        List<ModelDetails> modelDetails = modelDetailsMapper.selectList(new QueryWrapper<ModelDetails>().eq("model_info_id", modelInfo.getId()).eq("model_type", 1));
        if (null == modelDetails || modelDetails.isEmpty()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "模型映射关系"));
        }


        //业务验证、输出数据源和模型匹配校验
        List<TableInfo> tableInfos = tableInfoMapper.selectListByProeIdAndTableId(analysisInfo.getProcId(), analysisMessageModelConfSaveDto.getTableInfoId());
        if (null == tableInfos || tableInfos.isEmpty()) {
            throw new RRException("输出仓库不存在！");
        }
        //验证其输出字段是否一致
        List<TableFieldInfo> tableFieldInfos = tableFieldInfoMapper.selectList(new QueryWrapper<TableFieldInfo>().eq("table_id", analysisMessageModelConfSaveDto.getTableInfoId()));
        if (null == tableFieldInfos || tableFieldInfos.isEmpty()) {
            throw new RRException("输出仓库不存在,输出字段！");
        }
        List<ModelDetails> modelDetailsOutPut = modelDetailsMapper.selectList(new QueryWrapper<ModelDetails>().eq("model_type", 2).eq("model_info_id", modelInfo.getId()));
        if (null == modelDetailsOutPut || modelDetailsOutPut.isEmpty()) {
            throw new RRException("模型没有输出字段！");
        }
        //上线时放开
//        cheackModelField(tableFieldInfos,modelDetailsOutPut);

        AnalysisModelDataAndMappingDto analysisModelDataAndMappingDto = new AnalysisModelDataAndMappingDto();
        pingResultData(analysisModelDataAndMappingDto, analysisMessageModelConfSaveDto.getMessageData(), modelDetails);
        Map<String, Object> modelMap = analysisModelDataAndMappingDto.getModelMap();
        Map<String, Object> kafkaMap = analysisModelDataAndMappingDto.getKafkaMap();
        if (null == analysisMessageModelConfSaveDto.getAnalysisInfoConfigDtos() || analysisMessageModelConfSaveDto.getAnalysisInfoConfigDtos().isEmpty()) {
            //全部匹配，不需要进行模型映射
            if (!modelMap.isEmpty()) {
                throw new RRException("模型映射匹配不成功！");
            }
        } else {
            //验证模型字段和数据源字段是否有重复的
            cheackTypeName(analysisMessageModelConfSaveDto.getAnalysisInfoConfigDtos(), modelMap, kafkaMap);
        }

        //进行保存数据。先删除，再保存
        analysisMessageModelMapper.delete(new QueryWrapper<AnalysisMessageModel>().eq("analysis_info_id", analysisInfo.getId()));
        analysisMessageModelConfMapper.delete(new QueryWrapper<AnalysisMessageModelConf>().eq("analysis_info_id", analysisInfo.getId()));


        //更新对应的分析任务的配置
        analysisInfo.setMessageData(analysisMessageModelConfSaveDto.getMessageData());
        analysisInfo.setConnectId(analysisMessageModelConfSaveDto.getConnectId());
        analysisInfo.setModelInfoId(analysisMessageModelConfSaveDto.getModelInfoId());
        analysisInfo.setTableInfoId(analysisMessageModelConfSaveDto.getTableInfoId());
        analysisInfo.setOutputDataPartition(analysisMessageModelConfSaveDto.getOutputDataPartition());
        analysisInfo.setModTime(LocalDateTime.now());
        analysisInfo.setModPer(ShiroUtils.getUserId().intValue());
        analysisInfoMapper.updateById(analysisInfo);

        //保存数据
        if (null != analysisMessageModelConfSaveDto.getAnalysisInfoConfigDtos() && analysisMessageModelConfSaveDto.getAnalysisInfoConfigDtos().size() > 0) {
            analysisMessageModelConfSaveDto.getAnalysisInfoConfigDtos().forEach(analysisInfoConfigDto -> {
                AnalysisMessageModel analysisMessageModel = new AnalysisMessageModel();
                analysisMessageModel.setModelInfoId(analysisMessageModelConfSaveDto.getModelInfoId());
                analysisMessageModel.setAnalysisInfoId(analysisMessageModelConfSaveDto.getId());
                analysisMessageModel.setMessageField(analysisInfoConfigDto.getMessageField());
                //默认string
                analysisMessageModel.setMessageFieldType("String");
                analysisMessageModelMapper.insert(analysisMessageModel);

                AnalysisMessageModelConf analysisMessageModelConf = new AnalysisMessageModelConf();
                analysisMessageModelConf.setModelInfoId(analysisMessageModelConfSaveDto.getModelInfoId());
                analysisMessageModelConf.setAnalysisInfoId(analysisMessageModelConfSaveDto.getId());
                analysisMessageModelConf.setModelMappingId(analysisInfoConfigDto.getModelMappingId());
                analysisMessageModelConf.setMessageMappingId(analysisMessageModel.getId());
                analysisMessageModelConfMapper.insert(analysisMessageModelConf);

            });
        }


    }

    /**
     * @Description
     * @Author hujz
     * @Date 2019/9/3 18:02
     * @Param 验证模型的字段
     */
    @Override
    public void cheackModelField(List<TableFieldInfo> tableFieldInfos, List<ModelDetails> modelDetailsOutPut) {

        if (null == tableFieldInfos || tableFieldInfos.isEmpty()
                || null == modelDetailsOutPut || modelDetailsOutPut.isEmpty()) {
            return;
        }
        List<String> stringList = tableFieldInfos.stream().map(TableFieldInfo::getFieldName).collect(Collectors.toList());

        modelDetailsOutPut.forEach(modelDetail -> {
            if (!stringList.contains(modelDetail.getModelField())) {
                throw new RRException("模型的返回值和输出仓库的字段不一致！" + modelDetail.getModelField());
            }
        });

    }

    private void cheackTypeName(List<AnalysisInfoConfigDto> analysisInfoConfigDtos, Map<String, Object> modelMap, Map<String, Object> kafkaMap) {
        if (analysisInfoConfigDtos.isEmpty() || modelMap.isEmpty() || kafkaMap.isEmpty()) {
            return;
        }
        Set<String> messageFieldSet = new HashSet<>();
        Set<String> modelFieldSet = new HashSet<>();
        analysisInfoConfigDtos.forEach(analysisInfoConfigDto -> {
            if (StringUtils.isEmpty(analysisInfoConfigDto.getMessageField())
                    || StringUtils.isEmpty(analysisInfoConfigDto.getModelField())) {
                throw new RRException("模型映射匹配存在空的字段！");
            }
            if (null == analysisInfoConfigDto.getModelMappingId()) {
                throw new RRException("模型映射匹配中modelMappingId必须传！");
            }
            messageFieldSet.add(analysisInfoConfigDto.getMessageField());
            modelFieldSet.add(analysisInfoConfigDto.getModelField());
        });
        if ((messageFieldSet.size() != modelFieldSet.size())
                || (messageFieldSet.size() != analysisInfoConfigDtos.size())
                || (analysisInfoConfigDtos.size() != modelFieldSet.size())) {
            throw new RRException("模型映射匹配存在空的字段！");
        }
        //验证和数据库中的字段是否一致
        if (modelMap.size() != modelFieldSet.size()) {
            throw new RRException("未匹配的字段和实际的未匹配的字段不一致！");
        }
        modelFieldSet.forEach(str -> {
            if (!modelMap.containsKey(str)) {
                throw new RRException("未匹配的字段和实际的未匹配的字段不一致！");
            }
        });

    }

    /**
     * @Description 拼接数据对象
     * @Author hujz
     * @Date 2019/8/29 20:16
     * @Param analysisModelDataAndMappingDto：需要拼接数据的实体  msg：kafka数据
     */
    private void pingResultData(AnalysisModelDataAndMappingDto analysisModelDataAndMappingDto, String msg, List<ModelDetails> modelDetails) {

        if (null == analysisModelDataAndMappingDto) {
            analysisModelDataAndMappingDto = new AnalysisModelDataAndMappingDto();
        }
        if (StringUtils.isEmpty(msg) || null == modelDetails || modelDetails.isEmpty()) {
            return;
        }
        JSONObject jsonObject = JSONObject.parseObject(msg);
        Map<String, Object> resMap = jsonObject.getInnerMap();
        Map<String, Object> kafkaMap = new HashMap<>(resMap);
        Map<String, Object> modelMap = new HashMap<>();
        modelDetails.forEach(modelDetails1 -> {
            if (!resMap.containsKey(modelDetails1.getModelField())) {
                modelMap.put(modelDetails1.getModelField(), modelDetails1);
            } else {
                kafkaMap.remove(modelDetails1.getModelField());
            }
        });
        analysisModelDataAndMappingDto.setModelMap(modelMap);
        analysisModelDataAndMappingDto.setKafkaMap(kafkaMap);
        analysisModelDataAndMappingDto.setMessageData(msg);

    }


    /**
     * @Description 验证连接的参数
     * @Author hujz
     * @Date 2019/8/29 15:35
     * @Param
     */
    @Override
    public void cheackConfConnect(ConfConnect confConnect) {
        if (null == confConnect) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "连接"));
        }
        if (null == confConnect.getTopic() || confConnect.getTopic().isEmpty()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "topic"));
        }
        if (null == confConnect.getKafkaUrl() || confConnect.getKafkaUrl().isEmpty()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "kafkaUrl"));
        }
//        if (null == confConnect.getPort() || confConnect.getPort().isEmpty()) {
//            throw new RRException(String.format(BizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "port"));
//        }
    }
}
